package com.cch.bigdata.flink.ods

import com.cch.bigdata.flink.model.{AreaEvent, AreaInfo}
import com.alibaba.fastjson.{JSON, JSONArray, JSONObject}
import org.apache.hadoop.conf.Configuration
import org.apache.flink.configuration
import org.apache.flink.streaming.api.functions.sink.{RichSinkFunction, SinkFunction}
import org.apache.hadoop.hbase.{HBaseConfiguration, HConstants, TableName}
import org.apache.hadoop.hbase.client.{Connection, ConnectionFactory, Delete, Put, Table}
import org.apache.hadoop.hbase.util.Bytes

import java.{lang, util}

class HbaseSink extends RichSinkFunction[util.ArrayList[AreaEvent]]{

  val columnFamily:String = "area"

  var tableName = "lagou_area";
  var connection : Connection = _
  var hbaseTable : Table = _


  override def open(parameters: configuration.Configuration): Unit = {
    connection = getConnection()
    hbaseTable = connection.getTable(TableName.valueOf(tableName))
  }

  def processInsert(event: AreaInfo) = {
    val put = new Put(event.id.toString.getBytes)
    put.addColumn(columnFamily.getBytes(), "name".getBytes(), event.name.getBytes())
    put.addColumn(columnFamily.getBytes(), "pid".getBytes(), event.pid.toString.getBytes())
    put.addColumn(columnFamily.getBytes(), "sname".getBytes(), event.sname.getBytes())
    put.addColumn(columnFamily.getBytes(), "level".getBytes(), event.level.toString.getBytes())
    put.addColumn(columnFamily.getBytes(), "citycode".getBytes(), event.citycode.getBytes())
    put.addColumn(columnFamily.getBytes(), "yzcode".getBytes(), event.yzcode.getBytes())
    put.addColumn(columnFamily.getBytes(), "mername".getBytes(), event.mername.getBytes())
    put.addColumn(columnFamily.getBytes(), "lng".getBytes(), event.Lng.toString.getBytes())
    put.addColumn(columnFamily.getBytes(), "lat".getBytes(), event.Lat.toString.getBytes())
    put.addColumn(columnFamily.getBytes(), "pinyin".getBytes(), event.pinyin.getBytes())

    hbaseTable.put(put)
    println("数据插入成功")

  }

  def processUpdate(event: AreaInfo): Unit = {
    processInsert(event)//hbase会进行版本变更
  }

  def processDelete(event: AreaInfo): Unit = {
    val delete = new Delete(event.id.toString.getBytes())
    hbaseTable.delete(delete)
    println("删除成功")
  }

  def processDefault(event: AreaInfo): Unit = {

  }

  override def invoke(value: util.ArrayList[AreaEvent], context: SinkFunction.Context[_]): Unit = {
    value.forEach(event=>{
      val opType: String = event.opType
      val jsonData: String = event.opData
      val data: JSONObject = JSON.parseObject(jsonData)

      val id: Integer = data.getInteger("id");
      val name: String = data.getString("name")
      val pid: Integer = data.getInteger("pid")
      val sname: String = data.getString("sname")
      val level: Integer = data.getInteger("level")
      val cityCode: String = data.getString("citycode")
      val yzcode: String = data.getString("yzcode")
      val mername: String = data.getString("mername")
      val Lng: lang.Float = data.getFloat("Lng")
      val Lat: lang.Float = data.getFloat("Lat")
      val pinyin: String = data.getString("pinyin")
      val areaInfo:AreaInfo = AreaInfo(id, name, pid, sname, level, cityCode, yzcode, mername, Lng, Lat, pinyin)
      opType match{
        case "INSERT" => processInsert(areaInfo)
        case "UPDATE" => processUpdate(areaInfo)
        case "DELETE" => processDelete(areaInfo)
        case _ => processDefault(areaInfo)
      }
    })
  }


  override def close(): Unit = {
    if(hbaseTable!=null){
      hbaseTable.close()
    }

    if(connection!=null){
      connection.close()
    }
  }

  def getConnection():Connection ={
    val conf : Configuration = HBaseConfiguration.create()
    conf.set("hbase.zookeeper.quorum","localhost")
    conf.set("hbase.zookeeper.property.clientPort","2181")
    conf.setInt(HConstants.HBASE_CLIENT_OPERATION_TIMEOUT,30000)
    conf.setInt(HConstants.HBASE_CLIENT_SCANNER_TIMEOUT_PERIOD,30000)
    val connection = ConnectionFactory.createConnection(conf)
    connection
  }

}
